package com.cenntro.common.job;

//import com.dangdang.elasticjob.lite.annotation.ElasticSimpleJob;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.sql.DataSource;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;

import com.cenntro.common.mq.producer.MQProducerCondition;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.lite.api.listener.ElasticJobListener;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
//import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration.Builder;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;

@Configuration
//@ConditionalOnExpression("'${elaticjob.zookeeper.server-lists}'.length() > 0")
@Conditional(ElasticJobConfigCondition.class)
public class ElasticJobAutoConfiguration
{
  @Value("${elaticjob.zookeeper.server-lists}")
  private String serverList;
  @Value("${elaticjob.zookeeper.namespace}")
  private String namespace;
  @Autowired
  private ApplicationContext applicationContext;
  
  @PostConstruct
  public void initElasticJob()
  {
    ZookeeperRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration(this.serverList, this.namespace));
    regCenter.init();
    Map<String, SimpleJob> map = this.applicationContext.getBeansOfType(SimpleJob.class);
    for (Map.Entry<String, SimpleJob> entry : map.entrySet()) {
      SimpleJob simpleJob = (SimpleJob)entry.getValue();
      ElasticSimpleJob elasticSimpleJobAnnotation = (ElasticSimpleJob)simpleJob.getClass().getAnnotation(ElasticSimpleJob.class);
      
      String cron = (String)StringUtils.defaultIfBlank(elasticSimpleJobAnnotation.cron(), elasticSimpleJobAnnotation.value());
      SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(JobCoreConfiguration.newBuilder(simpleJob.getClass().getName(), cron, elasticSimpleJobAnnotation.shardingTotalCount()).shardingItemParameters(elasticSimpleJobAnnotation.shardingItemParameters()).failover(true).build(), simpleJob.getClass().getCanonicalName());
      LiteJobConfiguration liteJobConfiguration = LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build();
      
      String dataSourceRef = elasticSimpleJobAnnotation.dataSource();
      if (StringUtils.isNotBlank(dataSourceRef))   {
        if (!this.applicationContext.containsBean(dataSourceRef)) {
          throw new RuntimeException("not exist datasource [" + dataSourceRef + "] !");
        }
        DataSource dataSource = (DataSource)this.applicationContext.getBean(dataSourceRef);
        JobEventRdbConfiguration jobEventRdbConfiguration = new JobEventRdbConfiguration(dataSource);
        SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, jobEventRdbConfiguration, new ElasticJobListener[0]);
        jobScheduler.init();
      } else {
        SpringJobScheduler jobScheduler = new SpringJobScheduler(simpleJob, regCenter, liteJobConfiguration, new ElasticJobListener[0]);
        jobScheduler.init();
      }
    }
  }
}

